package com.bitcola.exchange.websocket;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.bitcola.exchange.biz.ColaExchangeBiz;
import com.bitcola.exchange.config.SpringBeanHandler;
import com.bitcola.exchange.message.NotifyMessage;
import lombok.extern.log4j.Log4j2;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;

@Component
@ServerEndpoint("/webSocket/v2")
@Log4j2
public class WebSocketService {

    /**
     * 订阅主题
     */
    public static Map<Session, List<String>> sessionTopics = new ConcurrentHashMap<>();

    /**
     * 容器
     */
    public static Map<String, ConcurrentHashMap<Session, Integer>> topicSession = new ConcurrentHashMap<>();

    /**
     * 服务线程数
     */
//    static ExecutorService service = Executors.newScheduledThreadPool(200);

    public static final String KLINE = "kline";

    public static final String SUBSCRIBE = "subscribe";

    public static final String TYPE = "type";

    public static final String PING = "ping";

    public static final String FROM = "from";

    public static final String TO = "to";

    /**
     * 请求
     */
    public static final String REQ = "req";

    public static final String OK = "ok";

    public static final String STATE = "subscribeState";

    /**
     * 全局数据
     */
    public static final String ALL = "_all";


    public static final String  MARKETALL="market_all";

    static Timer timer=new Timer();
    static {
        timer.schedule(new TimerTask() {
            @Override
            public void run() {

                for(Session session:sessionTopics.keySet()){

                    synchronized (session) {
                        if (!session.isOpen()) {
                            try {
                                session.close();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                            clearTopic(session);
                        }
                    }

                }
                log.info("清除已关闭session");
                log.info("session个数:"+sessionTopics.size());

            }
        },10000,60*1000);
    }

    @OnOpen
    public void onOpen(Session session) {
    }

    @OnError
    public void onError(Session session, Throwable error) {
        clearTopic(session);
    }

    /**
     * 连接关闭
     */
    @OnClose
    public void onClose(Session session) {
        clearTopic(session);
    }

    /**
     * 收到客户端的消息
     *
     * @param message 消息
     * @param session 会话
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        // 消息内容
        JSONObject jsonObject = JSONObject.parseObject(message);
        String type = jsonObject.getString( TYPE);
        if ( StrUtil.isEmpty( type)) {
            type = SUBSCRIBE;
        }
        switch (type) {
            case SUBSCRIBE: {
                var subscribes = jsonObject.getJSONArray( SUBSCRIBE);
                var sessionSubscribes = sessionTopics.getOrDefault(session, new ArrayList<>());
                for (String t : sessionSubscribes) {
                    var topicMap = topicSession.get(t);
                    if (t != null) topicMap.remove(session);

                }
                sessionSubscribes.clear();
                for (Object object : subscribes) {
                    String key = object.toString();
                    ConcurrentHashMap<Session, Integer> hashSet = topicSession.getOrDefault(key, new ConcurrentHashMap<>());
                    hashSet.put(session, 1);
                    sessionSubscribes.add(key);
                    if (key.contains(KLINE)){

                            this.sendFirstKline(key,session);

                    }else if(key.contains(MARKETALL)){
                            this.sendMarket(session);

                    }
                    topicSession.put(key, hashSet);
                }
                sessionTopics.put(session, sessionSubscribes);
                sendMessageToTopic(new NotifyMessage( STATE, OK));
                break;
            }
            case PING: {
                sendMessageToTopic(new NotifyMessage( PING,  OK));

                break;
            }
            case REQ: {
                Long from = jsonObject.getLong( FROM);

                Long to = jsonObject.getLong( TO);

                String key = jsonObject.getString( SUBSCRIBE);
                if ( ObjectUtil.isNotEmpty( from)  && ObjectUtil.isNotEmpty( to)) {

                        this.sendKline(from, to, key, session);

                }
                break;
            }
        }
    }

    /**
     * 发送历史kline数据
     * @param from
     * @param to
     * @param topic
     * @param session
     */
    public void sendKline(Long from, Long to, String topic, Session session){
        try {
            String [] split = StrUtil.split(topic, StrUtil.UNDERLINE);
            // 交易对
            String pair = StrUtil.concat( Boolean.FALSE, split[1], StrUtil.UNDERLINE, split[2]);
            // 分辨率
            String klineType = split[ 3];
            List<Number[]> kline = SpringBeanHandler.getBean( ColaExchangeBiz.class)
                    .kline(pair, from, to, 300, klineType);
            session.setMaxTextMessageBufferSize(100000);
            String jsonString = JSONObject.toJSONString(kline);
            // 开始发送消息
            sendByPage( jsonString, StrUtil.concat( Boolean.FALSE, topic, ALL), session);
        } catch (Exception e) {
            log.error("kline数据出错", e);
        }
    }

    /**
     * 第一次Kline数据
     * @param topic
     * @param session
     */
    public void sendFirstKline(String topic, Session session) {
        try {
            String [] split = StrUtil.split(topic, StrUtil.UNDERLINE);
            // 交易对
            String pair = StrUtil.concat( Boolean.FALSE, split[1], StrUtil.UNDERLINE, split[2]);
            // 分辨率
            String klineType = split[3];
            List<Number[]> kline = SpringBeanHandler.getBean( ColaExchangeBiz.class)
                    .kline(pair, null, null, 210, klineType);
            session.setMaxTextMessageBufferSize(100000);
            String jsonString = JSONObject.toJSONString(kline);
            // 开始发送消息
            sendByPage( jsonString, StrUtil.concat( Boolean.FALSE, topic, ALL), session);
        } catch (Exception e) {
            log.error("kline数据出错", e);
        }
    }


    public void sendMarket(Session session)
    {

        var markets=SpringBeanHandler.getBean( ColaExchangeBiz.class).getMarketAll();
        String jsonString = JSONObject.toJSONString(markets);
        sendByPage( jsonString, MARKETALL, session);

    }
    /**
     * 分页发送数据
     * @param sendStr
     * @param topic
     * @param session
     */
    public void sendByPage(String sendStr,String topic,Session session) {
        int length = sendStr.length();
        int count = 20000;
        int total = length%count>0?length/count+1:length/count;
        int page = 1;
        for (int i = 0; i < length; i+=count) {
            int subLength = i + count < length ? i + count : length;
            NotifyMessage notifyMessage = new NotifyMessage(topic,sendStr.substring(i,subLength));
            notifyMessage.setT(total);
            notifyMessage.setC(page);

                synchronized (session) {
                    if(session.isOpen())
                         session.getAsyncRemote().sendText(notifyMessage.toString());
                }

            page++;
        }
    }


    public void sendMessageToTopic(NotifyMessage message) {
        var sessions = topicSession.get( message.getTopic());
        if (sessions != null) {
            for (Session session : sessions.keySet()) {
                try {
                    synchronized (session){
                        if (session.isOpen()){
                            session.getAsyncRemote().sendText( message.toString());
                        }
                    }
                } catch (Exception e) {
                    log.error("===消息发送出错", e);
                }
            }
        }
    }

    /**
     * 移除订阅
     * @param session
     */
    public  static void clearTopic(Session session){
        var topics = sessionTopics.get(session);
        if (topics != null) {
            for (String topic : topics) {
                topicSession.get(topic).remove(session);
            }
            topics.clear();
            sessionTopics.remove(session);
        }
    }

}
